package es

import (
	"context"
	"errors"
	"gin-swagger-demo/pkg/logging"
	"github.com/olivere/elastic/v7"
	"log"
	"os"
	"sync"
)

var (
	esOnce   sync.Once
	esClient *elastic.Client
)

func GetEsClient() *elastic.Client {
	if esClient != nil {
		return esClient
	}
	esOnce.Do(func() {
		client, err := elastic.NewClient(
			elastic.SetURL("http://127.0.0.1:9200"),
			elastic.SetBasicAuth("", ""),
			elastic.SetErrorLog(log.New(os.Stderr, "", log.LstdFlags)),
			elastic.SetInfoLog(log.New(os.Stdout, "", log.LstdFlags)),
			//elastic.SetGzip(true), //启动gzip 压缩
			//elastic.SetHealthcheckInterval(10*time.Second), //设置监控检查时间间隔
			//elastic.SetSniff(false), //是否定期检查集群 默认true
			//...
		)
		if err != nil {
			logging.Error("GetEsClient err",err)
		}
		esClient = client
	})
	return esClient
}

// IndexExists 索引是否存在
func IndexExists(ctx context.Context, index string) (bool, error) {
	return GetEsClient().IndexExists(index).Do(ctx)
}

// CreateIndex 创建索引
func CreateIndex(ctx context.Context, index, desc string) error {
	exist, err := IndexExists(ctx, index)
	if err != nil {
		return err
	}
	if exist {
		return nil
	}
	_, err = GetEsClient().CreateIndex(index).BodyString(desc).Do(ctx)
	return err
}

// DeleteIndex 删除索引
func DeleteIndex(ctx context.Context, index string) error {
	exist, err := IndexExists(ctx, index)
	if err != nil {
		return err
	}
	if !exist {
		return nil
	}
	_, err = GetEsClient().DeleteIndex(index).Do(ctx)
	return err
}

// Create 创建单个文档
func Create(ctx context.Context, index, json string) error {
	_, err := GetEsClient().Index().Index(index).OpType("create").BodyJson(json).Refresh("true").Do(ctx)
	return err
}

// CreateWithId 创建单个文档 指定id
func CreateWithId(ctx context.Context, index, id, json string) error {
	_, err := GetEsClient().Index().Index(index).OpType("create").Id(id).BodyJson(json).Refresh("true").Do(ctx)
	return err
}

// CreateBulk 批量创建文档
func CreateBulk(ctx context.Context, index string, m map[string]string) error {
	bulkService := GetEsClient().Bulk().Index(index).Refresh("true")
	// 添加多个文档请求
	for id, doc := range m {
		bulkService.Add(elastic.NewBulkCreateRequest().
			Index(index).
			Id(id).
			Doc(doc))
	}
	res, err := bulkService.Do(ctx)
	if err != nil {
		return err
	}
	if len(res.Failed()) > 0 {
		return errors.New(res.Failed()[0].Error.Reason)
	}
	return nil
}

// Delete 通过 ID 删除文档
func Delete(ctx context.Context, index, id string) error {
	_, err := GetEsClient().Delete().Index(index).Id(id).Refresh("true").Do(ctx)
	return err
}

// DeleteBulk 删除文档批量（部分失败会返回失败）
// rets 删除的数量, error
func DeleteBulk(ctx context.Context, index string, ids []string) (int, error) {
	bulkService := GetEsClient().Bulk().Index(index).Refresh("true")
	for i := range ids {
		req := elastic.NewBulkDeleteRequest().Id(ids[i])
		bulkService.Add(req)
	}
	res, err := bulkService.Do(ctx)
	return len(res.Succeeded()), err
}

// Update 修改文档
// param: index 索引; id 文档ID; m 待更新的字段键值结构
func Update(ctx context.Context, index, id string, doc interface{}) error {
	_, err := GetEsClient().
		Update().
		Index(index).
		Id(id).
		Doc(doc).
		Refresh("true").
		Do(ctx)
	return err
}

// Upsert 修改文档（不存在则插入）
// params index 索引; id 文档ID; m 待更新的字段键值结构
func Upsert(ctx context.Context, index, id string, doc interface{}) error {
	_, err := GetEsClient().
		Update().
		Index(index).
		Id(id).
		Doc(doc).
		Upsert(doc).
		Refresh("true").
		Do(ctx)
	return err
}

// UpdateByQuery 根据条件修改文档
// param: index 索引; query 条件; script 脚本指定待更新的字段与值
func UpdateByQuery(ctx context.Context, index string, query elastic.Query, script *elastic.Script) (int64, error) {
	rsp, err := GetEsClient().
		UpdateByQuery(index).
		Query(query).
		Script(script).
		Refresh("true").
		Do(ctx)
	if err != nil {
		return 0, err
	}
	return rsp.Updated, nil
}

// UpdateBulk 批量修改文档
func UpdateBulk(ctx context.Context, index string, ids []string, docs []interface{}) error {
	bulkService := GetEsClient().Bulk().Index(index).Refresh("true")
	for i := range ids {
		doc := elastic.NewBulkUpdateRequest().
			Id(ids[i]).
			Doc(docs[i])
		bulkService.Add(doc)
	}
	res, err := bulkService.Do(ctx)
	if err != nil {
		return err
	}
	if len(res.Failed()) > 0 {
		return errors.New(res.Failed()[0].Error.Reason)
	}
	return nil
}

// UpsertBulk 批量修改文档（不存在则插入）
func UpsertBulk(ctx context.Context, index string, ids []string, docs []interface{}) error {
	bulkService := GetEsClient().Bulk().Index(index).Refresh("true")
	for i := range ids {
		doc := elastic.NewBulkUpdateRequest().
			Id(ids[i]).
			Doc(docs[i]).
			Upsert(docs[i])
		bulkService.Add(doc)
	}
	res, err := bulkService.Do(ctx)
	if err != nil {
		return err
	}
	if len(res.Failed()) > 0 {
		return errors.New(res.Failed()[0].Error.Reason)
	}
	return nil
}

// GetByID 根据ID查询单个文档
func GetByID(ctx context.Context, index, id string) (string, error) {
	res, err := GetEsClient().
		Get().
		Index(index).
		Id(id).
		Do(ctx)
	if err != nil {
		return "", err
	}
	return string(res.Source), nil
}

// IsDocExists 某条记录是否存在
func IsDocExists(ctx context.Context, id, index string) (bool, error) {
	return GetEsClient().Exists().Index(index).Id(id).Do(ctx)
}


// GetByQueryAll 根据条件查询
func GetByQueryAll(ctx context.Context, index string, query elastic.Query)([]string, error) {
	res, err := GetEsClient().Search(index).Query(query).Do(ctx)
	if err != nil {
		return nil, err
	}
	sl := make([]string, 0, res.TotalHits())
	for _, hit := range res.Hits.Hits {
		sl = append(sl, string(hit.Source))
	}
	return sl, nil
}

// GetByQueryPage 分页查询 (from + size 的优点是简单，缺点是在深度分页的场景下系统开销比较大。)
// param: index 索引; query 查询条件; page 起始页（从 1 开始）; size 页大小
func GetByQueryPage(ctx context.Context, index string, query elastic.Query, page, size int) ([]string, error) {
	start := (page - 1) * size
	res, err := GetEsClient().Search(index).Query(query).From(start).Size(size).Do(ctx)
	if err != nil {
		return nil, err
	}
	sl := make([]string, 0, res.TotalHits())
	for _, hit := range res.Hits.Hits {
		sl = append(sl, string(hit.Source))
	}
	return sl, nil
}

// GetByQueryPageSort 根据条件分页查询 & 指定字段排序
// param: index 索引; query 查询条件; page 起始页（从 1 开始）; size 页大小; field 排序字段; ascending 升序
// query := elastic.NewBoolQuery()
// query.Filter(elastic.NewRangeQuery("age").Gte(18))
// sl, err := GetByQueryPageSort(context.Background(), index, query, 1, 500, "create_time", false)
func GetByQueryPageSort(ctx context.Context, index string, query elastic.Query, page, size int, field string,
	ascending bool) ([]string, error) {
	from := (page - 1) * size
	res, err := GetEsClient().
		Search(index).
		Query(query).
		Sort(field, ascending).
		From(from).
		Size(size).
		Do(ctx)
	if err != nil {
		return nil, err
	}
	sl := make([]string, 0, res.TotalHits())
	for _, hit := range res.Hits.Hits {
		sl = append(sl, string(hit.Source))
	}
	return sl, nil
}



// GetByQueryPageSortScroll 获取第一页数据 & 获取游标ID (search after 可以实时高效的进行分页查询，但是它只能做下一页这样的查询场景，不能随机的指定页数查询。)
// ret: 文档切片, 游标ID, error
func GetByQueryPageSortScroll(ctx context.Context, index string, query elastic.Query, size int, field string,
	ascending bool) ([]string, string, error) {
	res, err := GetEsClient().
		Scroll(index).
		Query(query).
		Sort(field, ascending).
		Size(size).
		Do(ctx)
	if err != nil {
		return nil, "", err
	}
	sl := make([]string, 0, res.TotalHits())
	for _, hit := range res.Hits.Hits {
		sl = append(sl, string(hit.Source))
	}
	return sl, res.ScrollId, nil
}

// GetByQScrollID4ES 根据游标 ID 获取下一页
func GetByQScrollID(ctx context.Context, scrollID string) ([]string, error) {
	res, err := GetEsClient().
		Scroll().
		ScrollId(scrollID).
		Do(ctx)
	if err != nil {
		return nil, err
	}
	sl := make([]string, 0, res.TotalHits())
	for _, hit := range res.Hits.Hits {
		sl = append(sl, string(hit.Source))
	}
	return sl, nil
}
